package SparkStreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

/**
 * 基于HDFS文件的
 */
public class JavaHDFSWordCount {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("JavaSparkStreaming");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        //首先，使用JavaStreamingContext的textFileStream()方法，针对HDFS目录创建输入数据流
        JavaDStream<String> lines = jssc.textFileStream("hdfs://spark1:9000/wordcount_dir");
        JavaDStream<String> words = lines.flatMap(
                (FlatMapFunction<String, String>) s -> {
                    return null;
                    //return Arrays.asList(line.spilt(" "));
                }
        );

        JavaPairDStream<String, Integer> pairs = words.mapToPair(
                (PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1)
        );

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(
                (Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2
        );

        wordCounts.print();

        jssc.start();
        //jssc.awaitTermination();
        jssc.close();

    }
}
